package com.yyl.appx

import androidx.lifecycle.LifecycleCoroutineScope
import androidx.lifecycle.asLiveData
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.flow.*
import org.junit.Before
import org.junit.Test
import java.io.IOException

//https://juejin.cn/post/6989536876096913439
class FlowDemo {

    @Before
    fun testStart() {
//        冷流 🥶
//        冷指的是 无消费者时 则不会生产数据。
//        热流 🥵
//        热指的是 无观察者时 也会生产数据。
//    • 中间操作符
//    一般来说是用来执行一些操作，不会立即执行，返回值还是个Flow。
//    • 末端操作符
//    会触发流的执行，返回值不是Flow。
    }
    @Test
    fun test2() {
     //   testT3()
    }


    //创建Flow
    fun testT1() = runBlocking {
        // 一般的Fow , 仅有一个观察者 。冷流 。
        //构建
        val testFlow = flow<String> {
            emit("hello")
            emit("flow")
        }
        //接收
        testFlow.collect { value ->
            println("$value 1")
        }
        testFlow.collect { value ->
            println("$value 2")
        }
    }

    //    常用作与UI相关的数据观察，类比LiveData
    fun testT2() = runBlocking {
//        StateFlow
//        有状态的Flow ，可以有多个观察者，热流
//        构造时需要传入初始值 : initialState
//        常用作与UI相关的数据观察，类比LiveData
        //创建
        val uiState = MutableStateFlow(Result.Loading)
//监听
        uiState.collect { value ->
            println("$value 1")
        }
        uiState.collect { value ->
            println("$value 2")
        }
//赋值
        uiState.value = Result.Sucess
    }

    //    使用SharedFlow 你可以写个 FlowEventBus。
    fun testT3() = runBlocking {
//        SharedFlow   使用SharedFlow 你可以写个 FlowEventBus。
//        可定制化的StateFlow，可以有多个观察者，热流. 无需初始值，
//        构造参数有三个可选：
//        replay - 重播给新订阅者的值的数量（不能为负，默认为零）。
//        extraBufferCapacity - 除了replay之外缓冲的值的数量。当有剩余缓冲区空间时， emit不会挂起（可选，不能为负，默认为零）。
//        onBufferOverflow - 配置缓冲区溢出的操作（可选，默认为暂停尝试发出值）
//创建
        val signEvent = MutableSharedFlow<String>()

//监听
        signEvent.collect { value ->
            println(value)
        }
//赋值
        signEvent.tryEmit("hello")
        signEvent.tryEmit("shared flow")
    }

    fun testV1() = runBlocking {
//1>        创建Flow的基本方法.
        flow<Int> {
            emit(1)
            emit(2)
            emitAll(flowOf(4, 5, 6))
        }
//2>      快速创建 flow   集合向Flow的转换
        flowOf(1, 2, 3)
        listOf(1, 2, 3).asFlow()

        //3>        将回调方法改造成flow ,类似suspendCoroutine。
        fun flowFrom(): Flow<String> = callbackFlow {
            "asdfads"
        }
//4>        返回一个空流。
        emptyFlow<Int>()
//5>        在一般的flow在构造代码块中不允许切换线程，ChannelFlow则允许内部切换线程。
        val channelFlow = channelFlow<String> {
            send("hello")
            withContext(Dispatchers.IO) {
                send("channel flow")
            }
        }
    }

    //    末端操作符
    fun testM1() = runBlocking {
//    collect 触发flow的运行 。通常的监听方式。
        flowOf(1, 2, 3).collect { value ->
            print(value)
        }
//        collectIndexed 带下标的 收集操作
        flowOf(1, 2, 3).collectIndexed { index, value ->
            print("$index $value")
        }
//        collectLatest 与 collect的区别是有新值发出时，如果此时上个收集尚未完成，则会取消掉上个值的收集操作。
//        只想要最新的数据，中间值可以丢弃时可以使用此方式。
        flow {
            emit(1)
            delay(50)
            emit(2)
        }.collectLatest { value ->
            println("Collecting=$value - ")
            delay(100) // Emulate work
            println("$value collected")
            //输出// Collectin 1    Collecting 2     2 collected
        }
//        toCollection 将结果添加到集合。
//        toList 将结果转换为List。
//        toSet 将结果转换为Set。
        val array = arrayListOf(0)
        flow {
            emit(1)
            emit(2)
        }
//            .toList()
            .toCollection(array)
        array.forEach { value ->
            print(value)
        }
//        launchIn
//        直接触发流的执行，不设置action,入参为coroutineScope
        //        一般不会直接调用，会搭配别的操作符一起使用，如onEach,onCompletion 。返回值是Job。
//        flow {
//            emit(1)
//            emit(2)
//        }.launchIn ( lifecycleScope )
//        last 返回流 发出 的最后一个值 ,如果为空会抛异常。
//        lastOrNull 返回流 发出 的最后一个值 ,可以为空。
//        first 返回流 发出 的第一个值 ,如果为空会抛异常。
//        firstOrNull 返回流 发出 的第一个值 ,可以为空。
//        single 接收流发送的第一个值 ，区别于first(),如果为空或者发了不止一个值，则都会报错。
//        singleOrNull 接收流发送的第一个值 ，可以为空 ,发出多值的话除第一个，后面均被置为null。
//        count 返回流发送值的个数。类似 list.size() ，注：sharedFlow无效(无意义）
//        fold 从初始值开始 执行遍历,并将结果作为下个执行的 参数。
//        reduce 和fold 差不多，无初始值。
        val myFlow = flow {
            emit(1)
        }
        print(myFlow.singleOrNull()) // 1
        val sum = flowOf(2, 3, 4)
            .fold(10) { result, value ->
                println("fold=$result $value")
                result + value
            }
        println("sum=$sum")
        // sum = 19， 相当于 fold=10 2//fold=12 3//fold=15 4
        val result = flowOf(10, 2, 3)
            .reduce { acc, value ->
                println("reduce=$acc $value")
                acc + value
            }
        println("result=$result")
        //result=15 //reduce=10 2 //reduce=12 3
    }

    //    回调操作符
    fun testV2() = runBlocking {
//        onStart 在上游流开始之前被调用。可以发出额外元素,也可以处理其他事情，比如发埋点。
//        onCompletion 在流取消或者结束时调用。可以执行发送元素，发埋点等操作。
//        onEach 在上游向下游发出元素之前调用。
        flow<Int>{
            emit(1)
            emit(2)
            emit(3)
        }.onEach{ value->
            println(value)
        }
//        onEmpty 当流完成却没有发出任何元素时回调。可以用来兜底.。
//        onSubscription
//        SharedFlow 专属操作符 （StateFlow是SharedFlow 的一种特殊实现）
//        在建立订阅之后 回调。和 onStart 有些区别 ，SharedFlow 是热流，因此如果在onStart里发送值，则下游可能接收不到。
        val state = MutableSharedFlow<String>().onSubscription {
            emit("onSubscription")
        }
        state.collect { value ->
            println(value)
        }
    }

    //    变换操作符
    fun testV3() = runBlocking {
//        map 将发出的值 进行变换 ，lambda的返回值为最终发送的值。
//        mapLatest 类比 collectLatest ,当有新值发送时如果上个变换还没结束，会先取消掉。
//        mapNotNull 仅发送 map后不为空的值。
//        transform 对发出的值 进行变换 。区别于map， transform的接收者是FlowCollector
        flow {
            emit(1)
            emit(2)
        } .transform { value ->
            if (value == 1) {
                emit("value :$value*2")
            }
            emit("transform :$value")
        }.collect { value->
            println(value)
        }
//        transformLatest  类比 mapLatest ,当有新值发送时如果上个变换还没结束，会先取消掉。
//        transformWhile 这个变化的lambda 的返回值是 Boolean ,如果为 False则不再进行后续变换, 为 True则继续执行。
        flow {
            emit("a")
            emit("b")
        } .transformWhile { value ->
            emit(value)
            true
        } .collect { value->
            println(value)
        }
//        asStateFlow 将 MutableStateFlow 转换为 StateFlow ，就是变成不可变的。常用在对外暴露属性时使用。
//        asSharedFlow 将 MutableSharedFlow 转换为 SharedFlow ，就是变成不可变的。常用在对外暴露属性时使用。
        val _uiState = MutableStateFlow<Result>(Result.Loading)
        val uiState = _uiState.asStateFlow()
//        receiveAsFlow 将Channel 转换为Flow ,可以有多个观察者，但不是多播，可能会轮流收到值
        val _event = Channel<Result>()
        val event= _event.receiveAsFlow()
//        consumeAsFlow  将Channel 转换为Flow ,但不能多个观察者（会crash）!
        val _event2 = Channel<Result>()
        val event2= _event2.consumeAsFlow ()
//        withIndex  将结果包装成IndexedValue 类型。
        flow {
            emit("a")
            emit("b")
        } .withIndex().collect {
            print( "${it.index}: " + it.value)
        }
//        scan 和 fold 相似，区别是fold 返回的是最终结果，scan返回的是个flow ，会把初始值和每一步的操作结果发送出去。
        flowOf(1, 2, 3).scan(0) { acc, value ->
            acc + value
        }.collect {
            print(it)
        }
//        produceIn 转换为 ReceiveChannel , 不常用。
//        runningFold  区别于 fold ，就是返回一个新流，将每步的结果发射出去。
        flowOf(1, 2, 3).runningFold(1){ acc, value ->
            acc + value
        } .collect { value->
            print(value)
        }
//        runningReduce  区别于 reduce ，就是返回一个新流，将每步的结果发射出去。
        flowOf(1, 2, 3).runningReduce { acc, value ->
            acc + value
        } .collect { value->
            print(value)
        }
//        shareIn  将普通flow 转化为 SharedFlow , 其有三个参数：
        val share = flowOf(1,2,3).shareIn(this,SharingStarted.Eagerly)
//可以有多个观察者
        share.collect{value->
            print(value)
        }
    }

    //    过滤操作符
    fun test5() = runBlocking {
//        filter  筛选出符合条件的值。 filterNot相当于filter取反 filterNotNull 筛选不为空的值。
        flow {
            emit("a")
            emit("b")
        }.filter { value ->
            value == "a"
        }.collect { value->
            print(value)
        }
//        filterInstance 筛选对应类型的值。
        flow {
            emit("a")
            emit("b")
            emit(1)
        }.filterIsInstance<String>().collect { value->
            print(value)
        }
//        drop  入参count为int类型 ,作用是 丢弃掉前 n 个的值。
        flow {
            emit(1)
            emit(2)
            emit(3)
        }.drop(2).collect { value ->
            print(value)
        }
//        dropWhile
//        take 返回前 n个 元素。
        flow {
            emit(1)
            emit(2)
            emit(3)
        } .take(2) .collect { value ->
            print(value)
        }
//        debounce 防抖节流 ，指定时间内的值只接收最新的一个，其他的过滤掉。搜索联想场景适用。
        flow {
            emit(1)
            delay(90)
            emit(2)
            delay(90)
            emit(3)
            delay(1010)
            emit(4)
            delay(1010)
            emit(5)
        }.debounce(1000)
//        sample 采样 。给定一个时间周期，仅获取周期内最新发出的值。
        flow {
            repeat(10) {
                emit(it)
                delay(110)
            }
        }.sample(200)
//        distinctUntilChangedBy 去重操作符，判断连续的两个值是否重复，可以选择是否丢弃重复值。
        flowOf(
            Result.Loading,
            Result.Loading,
            Result.Sucess,
            Result.Sucess,
        ).distinctUntilChangedBy { it.name } .collect { value ->
            print(value.toString())
        }
//        distinctUntilChanged distinctUntilChangedBy 的简化调用
        flowOf(1, 1, 3,1).distinctUntilChanged()
            .collect { value ->
                print(value)
            }
    }

    //    组合操作符
    fun test6() = runBlocking {
//        combine 组合每个流最新发出的值。
        val flow = flowOf(1, 2).onEach { delay(10) }
        val flow2 = flowOf("a", "b", "c").onEach { delay(15) }
        flow.combine(flow2) { i, s -> i.toString() + s }.collect {
            println(it) // Will print "1a 2a 2b 2c"
        }
//        combineTransform 顾名思义 combine + transform**
        val numberFlow = flowOf(1, 2).onEach { delay(10) }
        val stringFlow = flowOf("a", "b", "c").onEach { delay(15) }
        numberFlow.combineTransform(stringFlow) { number, string ->
            emit("$number :$string")
        }.collect { value ->
            println(value)
        }
//        merge 合并多个流成 一个流。可以用在 多级缓存加载上。
        val numberFlow1 = flowOf(1, 2).onEach { delay(10) }
        val stringFlow2 = flowOf("a", "b", "c").onEach { delay(15) }

        listOf(numberFlow1, stringFlow2).merge()
            .collect { value ->
                print(value) //1a2bc
            }
//        flattenConcat 以顺序方式将给定的流展开为单个流 ，是Flow<Flow<T>>的扩展函数
        flow {
            emit(flowOf(1, 2, 3))
            emit(flowOf(4, 5, 6))
        }.flattenConcat().collect { value ->
            print(value)//123456
        }
//        flattenMerge 作用和 flattenConcat 一样，但是可以设置并发收集流的数量。
//        有个入参：concurrency: Int ,当其 == 1时，效果和 flattenConcat 一样，大于 1 时，则是并发收集。
        flow {
            emit(flowOf(1, 2, 3).flowOn(Dispatchers.IO))
            emit(flowOf(4, 5, 6).flowOn(Dispatchers.IO))
            emit(flowOf(7, 8, 9).flowOn(Dispatchers.IO))
        }.flattenMerge(3).collect { value ->
            print(value)
        }
//        flatMapContact 这是一个组合操作符，相当于 map + flattenConcat ,
//     通过 map 转成一个流，再通过flattenConcat展开合并成一个流。
        flowOf(1, 2, 3).flatMapConcat {
            flowOf("$it map")
        }.collect { value ->
            println(value)
        }
//        flatMapLatest 和其他 带 Latest的操作符 一样，如果下个值来了，上变换还没结束，就取消掉。
//        相当于 transformLatest + emitAll。
        flow {
            emit("a")
            delay(100)
            emit("b")
        }.flatMapLatest { value ->
            flow {
                emit(value)
                delay(200)
                emit(value + "_last")
            }
        }.collect { value ->
            print(value)
        }
//        flatMapMerge 也是组合操作符，简化使用。map + flattenMerge 。
//        因此也是有 concurrency: Int 这样一个参数，来限制并发数。
        flowOf("a", "b", "c", "d", "e", "f").flatMapMerge(3) { value ->
            flow {
                emit(value)
            }.flowOn(Dispatchers.IO)
        }.collect { value ->
            print(value)// b a c d e f
        }
//        zip 对两个流进行组合，分别从二者取值，一旦一个流结束了，那整个过程就结束了。
        val flow4 = flowOf(1, 2, 3).onEach { delay(10) }
        val flow5 = flowOf("a", "b", "c", "d").onEach { delay(15) }
        flow4.zip(flow5) { i, s -> i.toString() + s }.collect {
            println(it)
        }
    }

    //    功能性操作符
    fun test7() = runBlocking {
//cancellable
        // 接收的的时候判断 协程是否被取消 ，如果已取消，则抛出异常。
//        val job = flowOf(1, 3, 5, 7).cancellable().onEach { value ->
//            print(value)
//        }.launchIn(lifecycleScope)
//        //取消
//        job.cancel()

        //catch
        // 对上游异常进行捕获 ，对下游无影响 上游指此操作符之前的流
        flow<Int> {
            throw IOException("")
        }.catch { e ->
            if (e is IOException) {
                //...
            }
        }
//        retryWhen
//        有条件的进行重试 ，lambda 中有两个参数: 一个是 异常原因，一个是当前重试的 index (从0开始)。
//        lambda 的返回值 为 Boolean ，true则继续重试 ,false 则结束重试。
        flow<Int> {
            print("doing")
            throw IOException("")
        }.retryWhen { cause, attempt ->
            if (attempt > 4) {
                return@retryWhen false
            }
            cause is IOException
        }
//        retry
//        重试机制 ，当流发生异常时可以重新执行。retryWhen 的简化版。
//   参数Long    指定重试次数，以及控制是否继续重试。(返回值默认为true)
        flow<Int> {
            throw IOException("")
        }.retry(3) { e ->
            e is IOException //是否继续重试
        }
        flow<Int> {
            throw IOException("")
        }.retry(3)
//        buffer
//        如果操作符的代码需要相当****长时间来执行 ，可使用buffer操作符在执行期间为其创建一个单独的协程。
//        参数：
//        capacity: Int = BUFFERED 缓冲区的容量。
//        onBufferOverflow: BufferOverflow = BufferOverflow.``SUSPEND **溢出的话执行的操作。
//        有三个选择 ：SUSPEND 挂起， DROP_OLDEST 丢掉旧的，DROP_LATEST 丢掉新的。
        flowOf("A", "B", "C")
            .onEach { print("1$it - ") }
            .collect { print("2$it - ") }
        //        1A - 2A - 1B - 2B - 1C - 2C -
        println()
        flowOf("A", "B", "C")
            .onEach { print("1$it - ") }
            .buffer()
            .collect { print("2$it - ") }
//        1A - 1B - 1C - 2A - 2B - 2C -

//        conflate
//        仅保留最新值, 内部就是 buffer(CONFLATED)
        flow {
            repeat(30) {
                delay(100)
                emit(it)
            }
        }.conflate().onEach { delay(1000) }.collect { value ->
            print("$value-") //0-9-18-27-29-   结果不固定
        }
//        flowOn
//        指定上游操作的执行线程
        flowOf("A", "B", "C")
            .map { it } // Will be executed in IO
            .flowOn(Dispatchers.IO) // This one takes precedence
            .collect { }
    }
    @Test
    fun test() {
        testDemo()
    }
    fun testDemo() = runBlocking {
        val sum = flowOf(2, 3, 4)
            .fold(10) { result, value ->
                println("fold=$result $value")
                result + value
            }
        println("sum=$sum")
        // sum = 19， 相当于 fold=10 2//fold=12 3//fold=15 4
        val result = flowOf(10, 2, 3)
            .reduce { acc, value ->
                println("reduce=$acc $value")
                acc + value
            }
        println("result=$result")
        //result=15 //reduce=10 2 //reduce=12 3
    }


    enum class Result {
        Loading, Sucess
    }
}